package com.camemax.operators;

import com.camemax.pojo.LoginEvent;
import com.camemax.pojo.LoginFailedWarning;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.curator4.com.google.common.collect.Iterators;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;

public class LoginFailedProcessModify extends KeyedProcessFunction<Long, LoginEvent, LoginFailedWarning> {

    private ListState<LoginEvent> loginFailedEvents;

    @Override
    public void open(Configuration parameters) {
        loginFailedEvents = getRuntimeContext().getListState(
                new ListStateDescriptor<>("failed-login-history", LoginEvent.class)
        );
    }

    @Override
    public void processElement(LoginEvent value, Context ctx, Collector<LoginFailedWarning> out) throws Exception {
        if ("fail".equals(value.getLoginStatus())) {
            // 判断是否存在上一个"fail"状态数据
            Iterator<LoginEvent> iterator = loginFailedEvents.get().iterator();
            if (iterator.hasNext()) {
                LoginEvent loginFailed = iterator.next();
                if (value.getLoginDate() - loginFailed.getLoginDate() <= 2) {
                    out.collect(new LoginFailedWarning(
                            ctx.getCurrentKey(),
                            ctx.getCurrentKey() + " failed to login within 2s.",
                            LocalDateTime.ofEpochSecond( loginFailedEvents.get().iterator().next().getLoginDate()
                                    , 0
                                    , ZoneOffset.ofHours(8))
                                        .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss")
                                    ),
                            LocalDateTime.ofEpochSecond( value.getLoginDate()
                                    , 0
                                    , ZoneOffset.ofHours(8))
                                    .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss")
                                    )
                    ));
                }
            }else {
//            else {
//                    loginFailedEvents.clear();
//                    loginFailedEvents.add(value);
//                //if (Iterators.size(loginFailedEvents.get().iterator()) == 2) {
//                //   loginFailedEvents.clear();
//                //}
//            }
                loginFailedEvents.add(value);
            }
        } else {
            loginFailedEvents.clear();
        }
    }
}
